/*
 * 文件名：StreamDataParseHandler.java
 * 版权：亚信联创版权所有
 * 描述：TODO
 * 修改人：yuanze
 * 修改时间：2014年11月13日
 * 修改内容：TODO
 */
package com.ailk.bigdata.etl.realstream.server.protocal.aipr.netty.handler;

import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;

import java.net.InetAddress;
import java.util.Date;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.ailk.bigdata.etl.common.server.net.NettyUtil;
import com.ailk.bigdata.etl.realstream.server.model.RealStreamDataPack;
import com.ailk.bigdata.etl.realstream.server.model.ValueEvent;
import com.ailk.bigdata.etl.realstream.server.monitor.EventCounter;
import com.ailk.bigdata.etl.realstream.server.tools.RealStreamTool;
import com.lmax.disruptor.RingBuffer;

/**
 * @description TODO
 * @author [yuanze]
 * @version [版本号,2014年11月13日]
 * @see [相关类/方法]
 * @since [产品/模块版本]
 */
@Sharable
public class StreamDataAcceptHandler extends SimpleChannelInboundHandler<RealStreamDataPack>
{
	private static Logger logger = LoggerFactory.getLogger(StreamDataAcceptHandler.class);
	private EventCounter eventCount;
	private String remoteAdd = "";

	public StreamDataAcceptHandler(EventCounter eventCount)
	{
		this.eventCount = eventCount;
	}

	@Override
	public void channelActive(ChannelHandlerContext ctx) throws Exception
	{
		remoteAdd = NettyUtil.getClientAdd(ctx);
		logger.info("accept data stream from remote add [" + remoteAdd + "]");
		// 实时流端口认证，暂注释
		// if (!RealStreamTool.AUTHORISED_USER_INFO.containsKey(remoteAdd))
		// {
		// logger.info("not pass authentication");
		// ctx.writeAndFlush("not pass authentication").addListener(ChannelFutureListener.CLOSE);
		// }
		// else
		// {
		logger.info("receive message from IP[" + remoteAdd + "].");
		ctx.writeAndFlush("Welcome to " + InetAddress.getLocalHost().getHostName() + "!\r\n");
		ctx.writeAndFlush("It is " + new Date() + " now.\r\n");
		RealStreamTool.ACCEPTING_CHANNEL_CACHE.put(remoteAdd, ctx);
		ctx.fireChannelActive();
		// }
	}

	/**
	 * @see io.netty.channel.ChannelInboundHandlerAdapter#channelInactive(io.netty.channel.ChannelHandlerContext)
	 */
	@Override
	public void channelInactive(ChannelHandlerContext ctx) throws Exception
	{
		String address = ctx.channel().remoteAddress().toString();
		logger.info("数据接收端口监测到remote[" + address + "]关闭,处理缓存信息");
	}

	@Override
	protected void channelRead0(ChannelHandlerContext ctx, RealStreamDataPack dPack) throws Exception
	{
		eventCount.incrementEventReceivedCount();
		int dataType = dPack.getDataType();
		if (RealStreamTool.UNITCODE_ACCEPTED_DISRUPTOR.containsKey(dataType))
		{
			eventCount.incrementDatasAcceptedCount();
			eventCount.incrementEventReceivedByType(Integer.toString(dataType));
			RingBuffer<ValueEvent> ringBuffer = RealStreamTool.UNITCODE_ACCEPTED_DISRUPTOR.get(dataType);
			long next = ringBuffer.next();
			try
			{
				ValueEvent event = ringBuffer.get(next);
				event.setDataPackage(dPack);
			}
			finally
			{
				ringBuffer.publish(next);
			}
		}
		else
		{
			logger.warn("接口单元[" + dPack.getDataType() + "]未注册 不能进行处理........");
		}
	}

	@Override
	public void channelReadComplete(ChannelHandlerContext ctx) throws Exception
	{
		ctx.flush();
	}

	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception
	{
		logger.error(ctx.channel().remoteAddress().toString() + "receive message exception happened ,caused by " + cause.getMessage());
		ctx.close();
	}
}
